package com.lxl.hadoop.selfoutputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * @Author lixilin
 * @Date 2022/1/2 17:10
 * @Description
 */
public class MyRecordWriter extends RecordWriter<Text, NullWritable> {

    FSDataOutputStream fosA;
    FSDataOutputStream fosB;
    public MyRecordWriter(TaskAttemptContext job) {
        try {
            // 获取文件系统
            FileSystem fs = FileSystem.get(job.getConfiguration());
            // 创建输出流
             fosA = fs.create(new Path("D:\\bigdata\\input-tmp-data\\log\\a.log"));
             fosB = fs.create(new Path("D:\\bigdata\\input-tmp-data\\log\\other.log"));
        } catch (IOException e) {
        }
    }

    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        if (key.toString().contains("atguigu")) {
            fosA.write(key.toString().getBytes());
            fosA.writeChars("\r\n");
        } else {
            fosB.write(key.toString().getBytes());
            fosB.writeChars("\r\n");
        }

    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(fosA);
        IOUtils.closeStream(fosB);

    }
}
